package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author: xu
 * @desc: 独立访客UV的计算
 * 程  序：mockLog -> Nginx -> Logger.sh -> Kafka(ZK) -> BaseLogApp -> kafka -> UniqueVisitApp -> Kafka
 * 执行流程
 * - 运行模拟生成日志的jar
 * - 将模拟生成的日志数据发给nginx进行负载均衡
 * - nginx将请求转发给三台日志采集服务
 * - 三台日志采集服务，将数据发送到kafka的ods_base_log主题中
 * - BaseLogApp应用程序从ods_base_log中读取数据，进行（启动、曝光、页面）分流
 * - UniqueVisitorApp从dwd_page_log（页面）主题读取
 * - 对pv数据进行过滤，然后发送会Kafka中
 * >  按照mid进行分组
 * >  使用filter算子对数据进行过滤
 * >  在过滤的时候，使用状态变量记录上次访问日期
 */
public class UniqueVisitApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.基本环境准备
        // 1.1  准备本地测试流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 设置并行度
        env.setParallelism(1);
        // 1.3 设置Checkpoint
        // env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        // env.getCheckpointConfig().setCheckpointTimeout(60000);
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // env.setStateBackend(new FsStateBackend("hdfs://node1:8020/gmall/checkpoint/UniqueVisitApp"))
        // System.setProperty("HADOOP_USER_NAME", "root");

        // TODO 2.从kafka中读取数据
        String sourceTopic = "dwd_page_log";
        String groupId = "unique_visit_app_group";
        DataStreamSource<String> kafkaStringStream = env.addSource(MyKafkaUtil.getKafkaSource(sourceTopic, groupId));

        // TODO 3.对Kafka的字符串转换为JSON对象，然后按照设备id进行分组
        KeyedStream<JSONObject, String> keyedStream = kafkaStringStream
                .map(JSON::parseObject)
                .keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        // TODO 4.过滤得到UV
        SingleOutputStreamOperator<JSONObject> uniqueVisitStream = keyedStream.filter(
                new RichFilterFunction<JSONObject>() {
                    // 声明状态变量，用于存放上次访问日期
                    private transient ValueState<String> lastVisitDateState = null;
                    // 转换日期格式工具类
                    private FastDateFormat fdf = null;

                    /**
                     * 初始化状态,日期工具类
                     */
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> lastVisitDateStateDes = new ValueStateDescriptor<>("last-visit-date-state", String.class);
                        fdf = FastDateFormat.getInstance("yyyyMMdd");
                        // 因为我们统计的是日活DAU，所以状态数据只在当天有效，过了一天就可以失效掉
                        lastVisitDateStateDes.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        this.lastVisitDateState = getRuntimeContext().getState(lastVisitDateStateDes);
                    }

                    @Override
                    public boolean filter(JSONObject jsonObj) throws Exception {
                        // 判断当前页面是否从别的页面跳转过来的，如果有说明该次访问一定不是当日首次
                        String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                        if (lastPageId != null && lastPageId.length() > 0) {
                            return false;
                        }
                        // 获取状态日期
                        String lastVisitDate = lastVisitDateState.value();
                        // 获取当前访问时间
                        String curVisitDate = fdf.format(jsonObj.getLong("ts"));
                        // 用当前页面的访问时间和状态时间进行对比
                        if (StringUtils.isNotBlank(lastVisitDate) && lastVisitDate.equals(curVisitDate)) {
                            System.out.println("已访问：lastVisitDate-" + lastVisitDate + ",||logDate:" + curVisitDate);
                            return false;
                        } else {
                            System.out.println("未访问：lastVisitDate-" + lastVisitDate + ",||logDate:" + curVisitDate);
                            lastVisitDateState.update(curVisitDate);
                            return true;
                        }
                    }
                }
        );
        // uniqueVisitStream.print(">>>>>");

        // TODO 5.向kafka中写回dwm层，需要将json转换为String
        // json->string
        String sinkTopic = "dwm_unique_visit";
        uniqueVisitStream
                .map(JSONAware::toJSONString)
                .addSink(MyKafkaUtil.getKafkaSink(sinkTopic));

        env.execute(UniqueVisitApp.class.getSimpleName());
    }
}
